package MountainMq.listener;


import MountainMq.po.TransMessagePO;
import MountainMq.service.TransMessageService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

@Slf4j
// 使用ChannelAwareMessageListener注册的监听器，就要在config里面去绑定队列
public abstract class AbstractMessageListener implements ChannelAwareMessageListener {
    @Value("${mountainmq.reconsumeTimes}")
    public Integer resendTimes;

    @Autowired
    private TransMessageService transMessageService;

    // 具体业务我们需要实现了这个抽象类的实体类，然后重写该方法的流程
    public abstract void receiveMessage(Message message);

    // 模板方法模式，挺6的
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        TransMessagePO transMessagePO = transMessageService.messageReceiveReady(
                messageProperties.getMessageId(),
                messageProperties.getReceivedExchange(),
                messageProperties.getReceivedRoutingKey(),
                messageProperties.getConsumerQueue(),
                new String(message.getBody())
        );
        log.info("收到消息：{}，消费次数：{}",messageProperties.getMessageId(),transMessagePO.getSequence());

        try{
            // 执行业务
            receiveMessage(message);
            // 手动ACK
            channel.basicAck(deliveryTag,false);
            // 消息消费成功 直接删除该消息
            transMessageService.messageReceiveSuccess(messageProperties.getMessageId());
        }catch (Exception e){ // 业务执行异常
            log.error(e.getMessage(),e);
            if(transMessagePO.getSequence() >= resendTimes){ // 如果超过设定的重复消费次数，直接拒绝，最终打到死信队列
                channel.basicReject(deliveryTag,false);
            }else{ // 没有超过消费次数，继续消费
                // 延时消费，而且要有规律的按照指数级别 比如 2s 4s 8s
                long delayTime = ((long) Math.pow(2, transMessagePO.getSequence())) * 1000;
                // 睡眠
                Thread.sleep(delayTime);
                // Nack true保证重回队列
                channel.basicNack(deliveryTag,false,true);
            }
        }
    }
}
